package org.ala.tiktools.redis;

import org.ala.tiktools.logs.LoggerFactory;
import org.slf4j.Logger;
import redis.clients.jedis.JedisPubSub;

/**
 * 接subscrib的线程
 *
 * @author ala
 * @date 2024-11-28 15:47
 */
public class Subscriber extends Thread {


    static Logger log = LoggerFactory.getLogger("tiktools", "redis-subscrib");


    /**
     *  要监听的channels
     */
    protected String[] channels;
    /**
     *  回调
     */
    protected ICallback callback;
    /**
     *
     */
    protected JedisPubSub sub;
    /**
     * 线程执行标记
     */
    protected volatile boolean isRunning = false;


    private Subscriber() {
    }
    public static Subscriber build(String[] channels, ICallback callback) {
        Subscriber sub = new Subscriber();
        sub.channels = channels;
        sub.callback = callback;
        sub.sub = new JedisPubSub() {
            public void onMessage(String channel, String message) {
                try {
                    log.info("[Subscriber onMessage] channel:" + channel + " msg:" + message);
                    if (callback != null) {
                        callback.onMessage(channel, message);
                    }
                } catch (Throwable e) {
                    log.warn("[Subscriber onMessage] " + e.getMessage(), e);
                }
            }
            public void onSubscribe(String channel, int subscribedChannels) {
                log.info("[Subscriber onSubscribe] channel:" + channel);
            }
            public void onUnsubscribe(String channel, int subscribedChannels) {
                log.info("[Subscriber onUnsubscribe] channel:" + channel);
            }
        };
        return sub;
    }


    /**
     *  线程执行方法
     */
    @Override
    public void run() {
        while (isRunning) {
            try {
                sub.subscribe(channels);
            } catch (Throwable e) {
                log.warn("[Subscriber run] " + e.getMessage(), e);
            }
        }
    }

    /**
     *  开启监听
     */
    public void start() {
        if (!isRunning) {
            synchronized (this) {
                if (!isRunning) {
                    isRunning = true;
                    super.start();
                }
            }
        }
    }
    /**
     *  关闭监听
     */
    public void close() {
        isRunning = false;
        sub.unsubscribe(channels);
    }


    /**
     *  回调
     */
    public static interface ICallback {
        void onMessage(String channel, String msg);
    }
}
